package com.example.core.filter.router;

import com.example.common.config.Rule;
import com.example.common.enums.ResponseCode;
import com.example.common.exception.ConnectException;
import com.example.common.exception.ResponseException;
import com.example.core.ConfigLoader;
import com.example.core.context.GatewayContext;
import com.example.core.filter.Filter;
import com.example.core.filter.FilterAspect;
import com.example.core.helper.AsyncHttpHelper;
import com.example.core.helper.ResponseHelper;
import com.example.core.response.GatewayResponse;
import com.netflix.hystrix.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.asynchttpclient.Request;
import org.asynchttpclient.Response;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeoutException;

import static com.example.common.constants.FilterConst.*;

/**
 * @Author Peng Yisheng
 * @Date 2024/12/4 5:42
 * @Description 路由过滤器
 */
@Slf4j
@FilterAspect(id = ROUTER_FILTER_ID,
        name = ROUTER_FILTER_NAME,
        order = ROUTER_FILTER_ORDER)
public class RouterFilter implements Filter {
    // 路由过滤器里是具体的下游接口请求，把这里的请求打印到access日志文件中
    private static Logger accessLog = LoggerFactory.getLogger("accessLog");

    @Override
    public void doFilter(GatewayContext gatewayContext) throws Exception {

        // 首先获取熔断降级的配置
        Optional<Rule.HystrixConfig> hystrixConfig = getHystrixConfig(gatewayContext);
        // 如果存在熔断配置就走熔断降级的逻辑
        if (hystrixConfig.isPresent()) {
            log.info("熔断配置：{}", hystrixConfig.get());
            routeWithHystrix(gatewayContext, hystrixConfig);
        } else {
            route(gatewayContext, hystrixConfig);
        }

    }

    private static Optional<Rule.HystrixConfig> getHystrixConfig(GatewayContext gatewayContext) {
        Rule rule = gatewayContext.getRule();
        Optional<Rule.HystrixConfig> hystrixConfig =
                rule.getHystrixConfigs().stream().filter(c ->
                        StringUtils.equals(c.getPath(), gatewayContext.getRequest().getPath())).findFirst();
        return hystrixConfig;
    }

    /**
     * 根据提供的GatewayContext和Hystrix配置，执行路由操作，并在熔断时执行降级逻辑。
     * 熔断会发生在：
     * 当 Hystrix 命令的执行时间超过配置的超时时间。
     * 当 Hystrix 命令的执行出现异常或错误。
     * 当连续请求失败率达到配置的阈值。
     *
     * @param gatewayContext
     * @param hystrixConfig
     */
    private void routeWithHystrix(GatewayContext gatewayContext, Optional<Rule.HystrixConfig> hystrixConfig) {

        HystrixCommand.Setter setter = HystrixCommand.Setter
                // 进行分组，相同命令归类，便于监控和管理
                .withGroupKey(HystrixCommandGroupKey.Factory.asKey(gatewayContext.getUniqueId()))
                // 给请求路径设置唯一标识的 Hystrix命令
                .andCommandKey(HystrixCommandKey.Factory.asKey(gatewayContext.getRequest().getPath()))
                .andThreadPoolPropertiesDefaults(HystrixThreadPoolProperties.Setter()
                        // 核心线程池大小，决定了线程池中可以同时执行的最大并发任务数
                        .withCoreSize(hystrixConfig.get().getThreadCoreSize()))
                .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
                        // 设置命令的执行隔离策略。Hystrix 支持两种隔离策略：
                        // THREAD：使用独立的线程来执行命令。这是默认的隔离策略，适用于需要长时间运行或可能阻塞的操作。
                        // SEMAPHORE：使用信号量来限制并发执行的命令数量。这种策略不涉及线程切换，适用于快速、非阻塞的操作。
                        .withExecutionIsolationStrategy(HystrixCommandProperties.ExecutionIsolationStrategy.THREAD)
                        // 设置命令的超时时间（以毫秒为单位），在指定的时间内没有完成，Hystrix 将触发超时逻辑
                        .withExecutionTimeoutInMilliseconds(hystrixConfig.get().getTimeoutInMilliseconds())
                        // 设置是否在超时时中断线程。如果设置为 true，当命令超时时，Hystrix 会尝试中断正在执行的线程。这有助于尽快释放资源
                        .withExecutionIsolationThreadInterruptOnTimeout(true)
                        // 启用或禁用超时机制。如果设置为 false，Hystrix 将不会检查命令的执行时间，这里为true，则检查
                        .withExecutionTimeoutEnabled(true)
                        // 启用熔断器，默认为 true
                        .withCircuitBreakerEnabled(true)
                        // 触发熔断的最小请求数，默认为 20
                        .withCircuitBreakerRequestVolumeThreshold(4)
                        // 触发熔断的错误率阈值，默认为 50%
                        .withCircuitBreakerErrorThresholdPercentage(50)
                        // 熔断器处于打开状态的时间窗口长度，默认为 5000 毫秒（5 秒）
                        .withCircuitBreakerSleepWindowInMilliseconds(5000)
                        // 滚动统计的时间窗口长度，默认为 10000 毫秒（10 秒）
                        .withMetricsRollingStatisticalWindowInMilliseconds(20000)
                        // 滚动统计的时间窗口被划分为多少个桶，默认为 10
                        .withMetricsRollingStatisticalWindowBuckets(10));

        // 创建一个新的HystrixCommand对象，用于执行实际的路由操作。
        new HystrixCommand<Object>(setter) {
            @Override
            protected Object run() throws Exception {
                // 在Hystrix命令中执行路由操作，这是实际的业务逻辑。
                route(gatewayContext, hystrixConfig).get();
                return null;
            }

            @Override
            protected Object getFallback() {
                // 当熔断发生时，执行降级逻辑。
                // 检查是否是因为超时引发的熔断
//                boolean isResponseTimedOut = isResponseTimedOut();
                if (isResponseTimedOut()) {
                    // 这里设置你自己的超时异常处理逻辑
                    // 例如，返回一个自定义的错误响应
                    gatewayContext.setResponse(GatewayResponse.buildGatewayResponse(ResponseCode.GATEWAY_FALLBACK));
                } else {
                    // 其他类型的熔断处理
                    gatewayContext.setResponse(GatewayResponse.buildGatewayResponse(hystrixConfig.get().getFallbackResponse()));
                }
                gatewayContext.written();
                ResponseHelper.writeResponse(gatewayContext);
                return null;
            }
        }.execute(); // 执行Hystrix命令。
    }

    private CompletableFuture<Response> route(GatewayContext gatewayContext, Optional<Rule.HystrixConfig> hystrixConfig) {
        // 构建完整的下游请求：地址、路径
        Request request = gatewayContext.getRequest().build();
//        log.info("request: {}", request);
        // 这里暂时只实现http协议，后续可以提供不同协议的router过滤器
        CompletableFuture<Response> future = AsyncHttpHelper.getInstance().executeRequest(request);

        boolean whenComplete = ConfigLoader.getConfig().isWhenComplete();
        if (whenComplete) {
            future.whenComplete((response, throwable) -> {
                complete(request, response, throwable, gatewayContext, hystrixConfig);
            });
        } else {
            future.whenCompleteAsync((response, throwable) -> {
                complete(request, response, throwable, gatewayContext, hystrixConfig);
            });
        }

        return future;
    }

    private void complete(Request request,
                          Response response,
                          Throwable throwable,
                          GatewayContext gatewayContext,
                          Optional<Rule.HystrixConfig> hystrixConfig) {
        // 请求已经处理完毕 释放请求资源
        gatewayContext.releaseRequest();

        // 请求重试
        // 获取请求重试次数
        int currentRetryTimes = gatewayContext.getCurrentRetryTimes();
        int confRetryTimes = gatewayContext.getRule().getRetryConfig().getTimes();

        // 判断是否出现异常 如果是 进行重试
        // 熔断异常不重试
        if ((throwable instanceof TimeoutException || throwable instanceof IOException) &&
                currentRetryTimes < confRetryTimes && !hystrixConfig.isPresent()) {
            // 请求重试
            doRetry(gatewayContext);
            return;
        }

        try {
            // 正常请求
            if (Objects.isNull(throwable)) {
                gatewayContext.setResponse(GatewayResponse.buildGatewayResponse(response));
                return;
            }

            // 异常请求
            String url = request.getUrl();
            if (throwable instanceof TimeoutException) {
                log.warn("complete time out {}", url);
                gatewayContext.setThrowable(new ResponseException(ResponseCode.REQUEST_TIMEOUT));
                gatewayContext.setResponse(GatewayResponse.buildGatewayResponse(ResponseCode.REQUEST_TIMEOUT));
            } else {
                gatewayContext.setThrowable(new ConnectException(throwable,
                        gatewayContext.getUniqueId(), url, ResponseCode.HTTP_RESPONSE_ERROR));
                gatewayContext.setResponse(GatewayResponse.buildGatewayResponse(ResponseCode.HTTP_RESPONSE_ERROR));
            }

        } catch (Throwable t) {
            gatewayContext.setThrowable(new ResponseException(ResponseCode.INTERNAL_ERROR));
            gatewayContext.setResponse(GatewayResponse.buildGatewayResponse(ResponseCode.INTERNAL_ERROR));
            log.error("complete error", t);
        } finally {
            // 请求结束，写响应和释放资源
            gatewayContext.written();
            ResponseHelper.writeResponse(gatewayContext);

            // 打印请求日志
            accessLog.info("[{}ms] {} {} {} {} {} {}",
                    System.currentTimeMillis() - gatewayContext.getRequest().getBeginTime(),
                    gatewayContext.getRequest().getModifyHost(),
                    gatewayContext.getRequest().getUniqueId(),
                    gatewayContext.getRequest().getMethod(),
                    gatewayContext.getRequest().getPath(),
                    gatewayContext.getResponse().getHttpResponseStatus().code(),
                    gatewayContext.getResponse().getFutureResponse().getResponseBodyAsBytes().length);
        }
    }

    private void doRetry(GatewayContext gatewayContext) {
        gatewayContext.setCurrentRetryTimes(gatewayContext.getCurrentRetryTimes() + 1);
        System.out.println("current retry times: " + gatewayContext.getCurrentRetryTimes());
        try {
            // 调用当前路由过滤器方法再次进行请求重试
            // 目前是立即重试
            doFilter(gatewayContext);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

}
